package com.bootx.config;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.UUID;

/**
 * @author black
 */
@Component
public class OrderStreamConsumer {

    private final ObjectMapper objectMapper = new ObjectMapper();

    @Resource
    private RedisTemplate redisTemplate;

    private static final String STREAM_KEY = "mystream.orders";
    private static final String GROUP = "order-consumers";
    String consumerName = UUID.randomUUID().toString();
    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> orderListenerContainer(
            RedisConnectionFactory factory) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                        .pollTimeout(Duration.ofSeconds(2))
                        .build();
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =
                StreamMessageListenerContainer.create(factory, options);


        Consumer consumer = Consumer.from(GROUP, consumerName);

        container.receive(consumer,
                StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()),
                (MapRecord<String, String, String> message) -> {
                    String json = message.getValue().get("json");
                    OrderMsg msg = null;
                    try {
                        msg = objectMapper.readValue(json, OrderMsg.class);
                    } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                    }
                    System.out.println("Consume order: " + msg);
                    redisTemplate.opsForStream().acknowledge(GROUP, message);
                });

        container.start();
        return container;
    }

    private void handleOrder(OrderMsg msg) {
        System.out.println("Consume order: " + msg);
        // TODO: 业务逻辑
    }

    // 定义消息模型（可用 record 动态处理，无需强转）
    public static class OrderMsg {
        public String msg_id;
        public long ts;
        public String type;
        public Data data;
        public static class Data {
            public String orderId;
            public double amount;
            public String userId;
        }
        @Override public String toString() {
            return "OrderMsg{orderId=" + data.orderId + ", amount=" + data.amount + "}";
        }
    }
}
